package com.devilvan.customPartition;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class CustomPartitionMapper extends Mapper<LongWritable, Text,Text,FlowEntity> {
    private Text text;
    private FlowEntity flowEntity;

    /**
     * 初始化方法
     */
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        text = new Text();
        flowEntity = new FlowEntity();
    }

    /**
     * 1363157985066	13726230503	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com	游戏娱乐
     * 24	27	2481	24681
     * 200
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] fields = value.toString().split("\t");
        // 手机号码
        String msiSdn = fields[1];
        text.set(msiSdn);
        // Other
        Long reportTime = Long.valueOf(fields[0].trim());
        String apMac = fields[2];
        String acMac = fields[3];
        String host = fields[4];
        String siteType = fields[5];
        Long upPackNum = Long.valueOf(fields[6]);
        Long downPackNum = Long.valueOf(fields[7]);
        Long upPayLoad = Long.valueOf(fields[8]);
        Long downPayLoad = Long.valueOf(fields[9]);
        String httpStatus = fields[10];
        // FlowEntity
        flowEntity.setReportTime(reportTime);
        flowEntity.setMsiSdn(msiSdn);
        flowEntity.setApMac(apMac);
        flowEntity.setAcMac(acMac);
        flowEntity.setHost(host);
        flowEntity.setSiteType(siteType);
        flowEntity.setUpPackNum(upPackNum);
        flowEntity.setDownPackNum(downPackNum);
        flowEntity.setUpPayLoad(upPayLoad);
        flowEntity.setDownPayLoad(downPayLoad);
        flowEntity.setHttpStatus(httpStatus);
        context.write(text,flowEntity);
    }
}
